package cn.doitedu.demo3;

import cn.doitedu.demo1.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/11/18
 * @Desc: 学大数据，上多易教育
 * <p>
 * -- 营销策略11：营销规则
 * 针对年龄在 20~28 之间，性别为 male的用户，                         -- 静态画像条件
 * 当他发生一个行为：添加购物车 时，即给他推送一个满200-50的优惠活动信息   -- 触发条件
 * <p>
 * -- 营销策略12：营销规则
 * 针对年龄在 22~26 之间，性别为 female的用户，                     -- 静态画像条件
 * 当他发生一个行为：视频播放 时，即给他推送一个满100-50的优惠活动信息   -- 触发条件
 * <p>
 * -- 营销策略13：营销规则
 * 针对年龄在 20~40 之间，性别为 female的用户，                     -- 静态画像条件
 * 当他发生一个行为：广告点击 时，即给他推送一个满150-20的优惠活动信息   -- 触发条件
 * <p>
 * -- 营销策略21：营销规则
 * 针对月平均消费额>200，性别为 male的用户，                     -- 静态画像条件
 * 当他发生3次以上 点赞 行为后                                  -- 实时动态画像条件
 * 等他再发生 下单行为，即给他推送一个 xxxx 优惠活动信息           -- 触发条件
 * <p>
 * -- 营销策略22：营销规则
 * 针对月平均消费额>500，性别为 female的用户，                     -- 静态画像条件
 * 当他发生2次以上 收藏 行为后                                  -- 实时动态画像条件
 * 等他再发生 加购行为，即给他推送一个 xxxx 优惠活动信息           -- 触发条件
 **/

@Slf4j
public class Demo3 {
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);


        // 构造 mysql-cdc的 source，来实时监听规则元数据变更
        MySqlSource<String> metaCdcSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .databaseList("doit42") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("doit42.rule_meta") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        DataStreamSource<String> metaStream = env.fromSource(metaCdcSource, WatermarkStrategy.noWatermarks(), "meta");

        SingleOutputStreamOperator<RuleMetaBean> ruleMetaBeanStream = metaStream.map(new MapFunction<String, RuleMetaBean>() {
            @Override
            public RuleMetaBean map(String metaCdcJson) throws Exception {

                JSONObject jsonObject = JSON.parseObject(metaCdcJson);
                String op = jsonObject.getString("op");

                JSONObject dataObject;
                if (op.equals("d")) {
                    dataObject = jsonObject.getJSONObject("before");
                } else {
                    dataObject = jsonObject.getJSONObject("after");
                }

                // 取表字段
                int ruleId = dataObject.getIntValue("rule_id");
                int modelId = dataObject.getIntValue("model_id");
                String ruleParamJson = dataObject.getString("rule_param_json");

                return new RuleMetaBean(ruleId, modelId, ruleParamJson, op);
            }
        });

        // 广播规则元数据流
        MapStateDescriptor<String, String> bcStateDesc = new MapStateDescriptor<>("bc_state", String.class, String.class);
        BroadcastStream<RuleMetaBean> metaBeanBroadcastStream = ruleMetaBeanStream.broadcast(bcStateDesc);

        /* ------------------------------------ */
        /* ------------------------------------ */

        // 读kafka中行为数据
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd-events")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setGroupId("doit42-rtmk")
                .setClientIdPrefix("doit42-client-")
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();

        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");

        SingleOutputStreamOperator<EventBean> beanStream = stream.map(logJson -> JSON.parseObject(logJson, EventBean.class));

        KeyedStream<EventBean, Long> keyedStream = beanStream.keyBy(bean -> bean.getUser_id());


        /* ------------------------------------ */
        /* ----------连接  规则元数据广播流  和  行为数据主流  --------------- */
        BroadcastConnectedStream<EventBean, RuleMetaBean> connectStream = keyedStream.connect(metaBeanBroadcastStream);

        SingleOutputStreamOperator<String> messages = connectStream.process(new KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>() {

            HashMap<Integer, RuleCalculatorV3> calculatorPool = new HashMap<>();

            @Override
            public void processElement(EventBean eventBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {

                // 遍历运算机池中的每一个规则运算机对象，让它处理当前收到的行为数据
                for (Map.Entry<Integer, RuleCalculatorV3> entry : calculatorPool.entrySet()) {
                    RuleCalculatorV3 calculator = entry.getValue();
                    calculator.calculate(eventBean,out);
                }

            }

            @Override
            public void processBroadcastElement(RuleMetaBean metaBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>.Context ctx, Collector<String> out) throws Exception {

                String op = metaBean.getOp();
                int ruleId = metaBean.getRule_id();
                int modelId = metaBean.getModel_id();

                if (op.equals("d")) {
                    // 规则下线处理，从贵哦运算机池中移除对应的规则的运算机对象
                    calculatorPool.remove(ruleId);
                    log.info("下线了一个规则,规则id:{},模型id:{}",ruleId,modelId);

                } else {
                    // 规则更新、或注入新规则，都是插入新规则运算机对象到运算机池
                    RuleModel1_Calculator calculator = new RuleModel1_Calculator();
                    // 初始化 运算机对象
                    calculator.init(getRuntimeContext(),metaBean.getRule_param_json());
                    // 将初始化好的运算机对象，放入运算机池
                    calculatorPool.put(ruleId,calculator);

                    log.info("注入了一个规则,规则id:{},模型id:{}",ruleId,modelId);

                }
            }
        });


        messages.print();

        env.execute();


    }
}
